package io.wongxd.demo

import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.MutableSharedFlow
import kotlinx.coroutines.flow.asFlow
import kotlinx.coroutines.flow.conflate
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.flow.launchIn
import kotlinx.coroutines.flow.onEach
import kotlinx.coroutines.flow.onStart
import kotlinx.coroutines.flow.produceIn
import kotlinx.coroutines.flow.receiveAsFlow
import kotlinx.coroutines.flow.transform
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.withContext
import org.junit.Test
import java.text.SimpleDateFormat

class FlowTest {

    fun Long.formatTime(): String {
        val sdf = SimpleDateFormat("HH:mm:ss")
        return sdf.format(this)
    }

    /** 外冷内热 **/
    fun createFlow() = flow<Long> {
        // 声明一个 Channel，当数据变化时，让 Flow emit data
        val observerChannel = Channel<Unit>(Channel.CONFLATED)
        // 首次创建 flow，一定会返回一次 query 的数据
        observerChannel.trySend(Unit)
        withContext(Dispatchers.IO) {
            // 这里会有 suspend 标记，等调用 observerChannel.offer() 会 resume
            // 之后再 suspend，如此往复
            for (signal in observerChannel) {

            }
        }
    }

    fun <T> createHotInsideFlow(firstEmit: suspend () -> T, emitter: (observerChannel: Channel<T>) -> Unit) = flow {
        // 声明一个 Channel，当数据变化时，让 Flow emit data
        val observerChannel = Channel<T>(Channel.CONFLATED)
        // 首次创建 flow，返回一次数据
        observerChannel.trySend(firstEmit())
        emitter.invoke(observerChannel)
        // 这里会有 suspend 标记，等调用 observerChannel.offer() 会 resume 之后再 suspend，如此往复
        for (signal in observerChannel) {
            emit(signal)
        }
    }

    @Test
    fun testF() = runBlocking {
        println("之：${1 shl 3}")
        launch(Dispatchers.Default) {
            createHotInsideFlow<String>(firstEmit = {
                "第一次 :${System.currentTimeMillis().formatTime()}"
            }) { observerChannel ->
                launch {
                    repeat(12) {
                        delay(1000)
                        val data = "后续发送 $it :${System.currentTimeMillis().formatTime()}"
                        println("发值:$data in io ${Thread.currentThread().name}")
                        withContext(Dispatchers.IO) {
                            observerChannel.trySend(data)
                        }
                    }
                }
            }.collect {
                println("取值:${it} collect ${Thread.currentThread().name}")
            }
        }.join()
        println(" done ")
    }

    @Test
    fun `test flow`() = runBlocking {
        println("之：${1 shl 3}")

        val sharedFlow = MutableSharedFlow<Int>(replay = 1)

        launch(Dispatchers.Default) {
            repeat(10) {
                delay(5)
                sharedFlow.emit(it)
                println("发值:$it in io ${Thread.currentThread().name}")
            }
        }

        sharedFlow
            .onStart {
                println("取值: onStart ${Thread.currentThread().name}")
            }
            .onEach {
                delay(150)
                println("取值:$it onEach ${Thread.currentThread().name}")
            }
            .conflate()
            .collect {
                withContext(Dispatchers.IO) {
                    delay(100)
                    println("取值值:$it in io ${Thread.currentThread().name}")
                }
                delay(100)
                println("取值值:$it  ${Thread.currentThread().name}")
            }

        println(" done ")
    }

    @Test
    fun `test channel 2 flow`() = runBlocking {
        val channel = Channel<String>()
        val flow = channel.receiveAsFlow()
        launch {
            flow
                .conflate()
                .collect {
                    println("collect:$it")
                    delay(500)
                }
        }

        delay(200)

        launch {
            repeat(100) {
                channel.send("hello fish  $it")
                delay(50)
            }

            channel.close()
            println(" channel.close ")
        }

        println(" done ")
    }


    @Test
    fun `test flow 2 channel`() = runBlocking {
        val flow = flow {
            repeat(100) {
                emit("hello fish $it")
                delay(100)
            }
        }
        val channel = flow.produceIn(this)

        launch {
            for (c in channel) {
                println(" info :$c ")
            }
        }

        println(" done ")
    }

    fun flagsSimple(flag: String): List<String> {
        return flag.split(",").map { it.trim() }
    }

    @Test
    fun `test other`() {
        println("测测: ${flagsSimple("body")}")

        println("测测: ${flagsSimple("shoes,")}")

        println("测测: ${flagsSimple("eye,hair")}")
    }

    @Test
    fun `test transform`() = runBlocking {
        (1..10).asFlow()
            .transform {
                if (it in 8..9) println("测测:等级1：$it")
                else emit(it)
            }
            .transform {
                if (it in 4..5) println("测测:等级2：$it")
                else emit(it)
            }
            .onEach {
                println("测测:等级3：$it")
            }
            .launchIn(this)

        println(" done ")
    }


}